/*
 * file_container.hpp
 *
 *  Created on: Apr 12, 2013
 *      Author: nick
 */

#ifndef FILE_CONTAINER_HPP_
#define FILE_CONTAINER_HPP_

#include <sys/stat.h>
#include <string>
#include <iostream>
#include <boost/iostreams/device/mapped_file.hpp>
#include "../hpp/converter.hpp"
#include "../hpp/sorter.hpp"

const int MEMORY_LIMIT_IN_PAGES = 2;
const long int MEMORY_LIMIT_IN_BYTES = MEMORY_LIMIT_IN_PAGES*boost::iostreams::mapped_file::alignment();

class Chunk {
private:
	const char* fileName;
	boost::iostreams::mapped_file_source source;
	boost::iostreams::mapped_file_sink sink;
	int id;
	bool first;
	bool last;
	__off_t offset;
	__off_t currentPosition;
private:
	std::string::const_iterator getFirstElement(const std::string & inputString) const {
		if(first)
			return inputString.begin();
		for(std::string::const_iterator chunkBegin = inputString.begin(); chunkBegin != inputString.begin() + boost::iostreams::mapped_file_source::alignment(); chunkBegin++) {
			if(*chunkBegin == ',')
				return ++chunkBegin;
			if(chunkBegin == inputString.end())
				return chunkBegin;
		}
		throw ParseException("Error from start : element is equal or larger than one page");
	}

	std::string::const_iterator getLastElement(const std::string & inputString) {
		if(!last) {
			for(std::string::const_iterator chunkEnd = inputString.begin() + (MEMORY_LIMIT_IN_BYTES - boost::iostreams::mapped_file_source::alignment()); chunkEnd != inputString.end(); chunkEnd++) {
				if(*chunkEnd == ',')
					return chunkEnd;
			}
		}
		return inputString.end();
	}
public:
	Chunk(const char* fileName, int id, __off_t offset) : fileName(fileName), id(id), first(false), last(false), offset(offset), currentPosition(0) {}

	void setFirst(bool first) {
		this->first = first;
	}

	void setLast(bool last) {
		this->last = last;
	}

	int getId() const {
		return id;
	}

	void process(ChunkConverter * converter, Sorter * sorter) {
		source.open(fileName, MEMORY_LIMIT_IN_BYTES, offset);
		std::string inputString(source.data());
		std::vector<DataType> result;
		std::string::const_iterator begin = getFirstElement(inputString);
		std::string::const_iterator end = getLastElement(inputString);
		if((std::distance(begin,end)) != 0) {
			converter->extract(begin, end, result);
			sorter->sort(result);
			sink.open(getParamsForOutput(std::distance(begin,end)));
			converter->collect(result, sink.data());
			sink.close();
		}
		source.close();
	}



	boost::iostreams::mapped_file_params getParamsForOutput(size_t size) const {
		std::ostringstream out;
		out << id;
		boost::iostreams::mapped_file_params params(out.str());
		params.new_file_size = size;
		return params;
	}

	~Chunk() {
		if(source.is_open())
			source.close();
		if(sink.is_open())
			sink.close();
	}

};

class ChunkContainer {
	std::vector<Chunk> chunks;
	const char * outputFileName;
private:
	ChunkContainer(const ChunkContainer & anotherChunkContainer);
	ChunkContainer & operator=(const ChunkContainer & anotherChunkContainer);
public:
	ChunkContainer(const char* inputFileName, const char * outputFileName) : outputFileName(outputFileName) {
		if(MEMORY_LIMIT_IN_PAGES <= 1) {
			Logging::fatal("Minimum memory limit for mapping - two pages. Exiting...");
		} else {
			createChunks(inputFileName);
		}
	}

	void createChunks(const char* fileName) {
		struct stat64 info;
		if(stat64(fileName, &info) == -1) {
			throw IOException(std::string("Failed to retrieve file size. {fileName : ") + fileName + "}");
		}
		__off_t size = info.st_size;
		int id = 0;
		for(__off_t start = 0; start < size; id++, start += MEMORY_LIMIT_IN_BYTES - boost::iostreams::mapped_file_source::alignment()) {
			chunks.push_back(Chunk(fileName, id, start));
		}
		if(!chunks.empty()) {
			Chunk & firstChunk = chunks.front();
			firstChunk.setFirst(true);
			Chunk & lastChunk = chunks.back();
			lastChunk.setLast(true);
		}
	}

	void process(ChunkConverter * converter, Sorter* sorter) {
		for(std::vector<Chunk>::iterator current = chunks.begin(); current != chunks.end(); current++) {
			(*current).process(converter, sorter);
		}


//---------------------------------------------------------------------------------------------------------------------------------------
//This is the worst code for merge stage, I must rewrite it
//This is merge stage, it is temporary solution(All files are mapped simultaneously and writing via streams)
//May be I am influenced by Java experience writing this code
		std::vector<boost::iostreams::mapped_file_source> mappedFiles;
		for(std::vector<Chunk>::iterator current = chunks.begin(); current != chunks.end(); current++) {
			std::ostringstream tmp;
			tmp << current->getId();
			mappedFiles.push_back(boost::iostreams::mapped_file_source(tmp.str()));
		}
		std::vector<std::vector<DataType> > arrays;
		for(std::vector<boost::iostreams::mapped_file_source>::iterator current = mappedFiles.begin(); current != mappedFiles.end(); current++) {
			std::vector<DataType> array;
			std::string inputString(current->data());
			converter->extract(inputString.begin(), inputString.end(), array);
			arrays.push_back(array);
		}
		std::ofstream out(outputFileName);
		while(!isAllArraysEmpty(arrays)) {
			std::vector<std::vector<DataType> >::iterator arrayWithMinimalFirstElement = getArrayWithMinimaFirstElement(arrays);
			out << arrayWithMinimalFirstElement->front();
			arrayWithMinimalFirstElement->erase(arrayWithMinimalFirstElement->begin());
			if(!isAllArraysEmpty(arrays)) {
				out << ',';
			}
		}
		out.close();
		for(std::vector<boost::iostreams::mapped_file_source>::iterator current = mappedFiles.begin(); current != mappedFiles.end(); current++) {
			current->close();
		}
	}

	std::vector<std::vector<DataType> >::iterator getArrayWithMinimaFirstElement(std::vector<std::vector<DataType> > & arrays) {
		std::vector<std::vector<DataType> >::iterator min;
		for(std::vector<std::vector<DataType> >::iterator current = arrays.begin(); current != arrays.end(); current++) {
			if(!current->empty())
				min = current;
		}
		for(std::vector<std::vector<DataType> >::iterator current = arrays.begin(); current != arrays.end(); current++) {
			if(!current->empty() && min->front() > current->front()) {
				min = current;
			}
		}
		return min;
	}

	bool isAllArraysEmpty(std::vector<std::vector<DataType> > & arrays) {
		for(std::vector<std::vector<DataType> >::iterator current = arrays.begin(); current != arrays.end(); current++) {
			if(!current->empty())
				return false;
		}
		return true;
	}
//------------------------------------------------------------------------------------------------------------------------------------------
};
#endif /* FILE_CONTAINER_HPP_ */
